#!/usr/bin/evn python
# -*- coding: utf-8 -*-
"""
-------------------------------------------------
   文件名称 :     serve
   文件功能描述 :   功能描述
   创建人 :       小钟同学
   创建时间 :          2021/12/6
-------------------------------------------------
   修改描述-2021/12/6:         
-------------------------------------------------
"""
import asyncio
import warnings
from functools import lru_cache
import sys
import time
import grpc
from concurrent import futures
import signal
import logging
import typing
import platform
import inspect
import multiprocessing
from multiprocessing import freeze_support
from dataclasses import dataclass
from grpcframe.zyxgrpc.base_middleware import BaseMiddleware
from grpcframe.zyxgrpc.logger import ConsoleLogger
from grpcframe.utils.signal_wrapper import signal_send
from grpcframe.config.serve_conf import AppSettings
from grpcframe.exts.register.base_register import Register
from grpcframe.exts import BaseExpand

from typing import TypeVar
from types import MethodType, FunctionType
from typing import ClassVar
from grpcframe.exts.grpclocal.ctx import RequestContext
import contextlib

_LOGGER = logging.getLogger(__name__)

# 自定义一种类型声明，方便做相关的类型的说明注解
# ClassVar--用于标记类变量的特殊类型构造
# 用于表示容器元素的预期类型
servericType = TypeVar('servericType', typing.Callable, typing.Sequence[typing.Callable])
# 自定义的类型
middlewareType = TypeVar('middlewareType', object, typing.Sequence[typing.Callable])


# 自定义内嵌的storeforstate
class State(object):
    """
    An object that can be used to store arbitrary state.

    Used for `request.state` and `app.state`.
    """

    def __init__(self, state: typing.Dict = None):
        if state is None:
            state = {}
        super(State, self).__setattr__("_state", state)

    def __setattr__(self, key: typing.Any, value: typing.Any) -> None:
        self._state[key] = value

    def __getattr__(self, key: typing.Any) -> typing.Any:
        try:
            return self._state[key]
        except KeyError:
            message = "'{}' object has no attribute '{}'"
            raise AttributeError(message.format(self.__class__.__name__, key))

    def setattr(self, key: typing.Any, value: typing.Any) -> None:
        self._state[key] = value


class FastGrpcApp:

    # 实例化配置信息
    def __init__(self,
                 host='127.0.0.1',
                 # 启动的服务的端口号
                 port=None,
                 # 工作线程数
                 max_workers=10,
                 # 是否开启热重启机制
                 re_load=False,
                 # 是否进行debug日志输出
                 debug: bool = False,
                 # 当前服务的版本信息
                 version: str = "0.1.0",
                 # 启动的时候是否检测有服务的列表已经被注册
                 is_check_servicers=True,
                 # 服务队列列表
                 servicers: typing.Sequence[typing.Callable] = None,
                 # 当前服务定义中间件列表
                 middlewares: typing.Sequence[BaseMiddleware] = None,
                 # 当前服务定义中间件列表
                 expands: typing.Sequence[BaseExpand] = None,
                 # 当前服务自定义异常处理器
                 exception_handlers=None,
                 # 当前服务自定义的回调处理事件
                 # 某些情况下，我们可能并不需要严格区分一个变量或参数到底是列表 list 类型还是元组 tuple 类型，我们可以使用一个更为泛化的类型，叫做 Sequence
                 on_startup: typing.Sequence[typing.Callable] = None,
                 on_shutdown: typing.Sequence[typing.Callable] = None,
                 # SERVER_PRIVATE_KEY
                 is_certificate=False,
                 serve_private_key=None,
                 server_certificate_chain=None,
                 register_center: typing.ClassVar[Register] = None,
                 is_tracer=False,
                 is_async=False
                 ):

        # app保持一个状态信息
        self.state: State = State()
        # 是否是使用异步协程的方式启动服务
        self.is_async = is_async
        # 异步循环事件
        self._loop = None
        # grpc服务对象
        self.fastgrpc_server = None
        # 当前版本 信息
        self._version = version
        # 是否debug
        self.debug = debug
        # 是否开启链路追踪中间件
        self.is_tracer = is_tracer
        # 配置当前工作线程数
        self._max_thread_pool_workers = max_workers
        # 配置进程延迟结束的间隔休眠时间
        self.grpc_grace = 1
        # 应用是否已经停止
        self._stopped = False
        # 服务配置信息读取
        self.config = self.get_app_settings()
        # 实例化日志对象
        self.logger = ConsoleLogger().logger
        # 配置当前的异常错误的处理
        self.exception_handlers = ({} if exception_handlers is None else dict(exception_handlers))
        # 配置当前的所有服务中间件信息
        self.user_middlewares = [] if middlewares is None else list(middlewares)
        # 当前用户自定义的扩展插件集对象
        self.user_expands = expands if expands is not None else {}
        # 定义事件回调列表
        self.on_startup = [] if on_startup is None else list(on_startup)
        self.on_shutdown = [] if on_shutdown is None else list(on_shutdown)
        # 是否开启热重启机制
        self.re_load = self._get_reload_monitor() if re_load else False
        # 是否启动前检测服务列表信息
        self.is_check_servicers = is_check_servicers
        # 初始化服务列表
        self.user_servicers = servicers if servicers is not None else {}
        # 初始化端口的服务号
        self.port = self._get_open_port() if port is None else port
        # 初始化服务地址
        self._address_full = host + ':' + str(self.port)
        # 单纯是HOST地址
        self._address_host = host
        # 是否启用SSL的认证
        self.is_certificate = is_certificate
        # 下面的方式可以继续优化
        self.server_private_key = serve_private_key
        self.server_certificate_chain = server_certificate_chain
        self.register_center = register_center

        # Store all of the services
        self._services: typing.Dict[str, "Service"] = {}


    @property
    def class_name(self):
        return self.__class__.__name__

    @property
    def address(self):
        return self._address_full

    @address.setter
    def address(self, value):
        self._address_full = value

    @property
    def version(self):
        return self._version

    @version.setter
    def version(self, value):
        self._version = value

    def request_context(self, request, context):
        """ 使用类似falask的自定义的方式封装的-请求上下文
        """
        return RequestContext(request, context)

    def get_certificate_chain(self):
        # 判断设置的安全秘钥信息
        assert self.server_private_key is not None or self.server_certificate_chain is not None
        # 开始读取相关信息
        if self.server_private_key and self.server_certificate_chain:
            with open(self.server_private_key, 'rb') as f:
                private_key = f.read()
            with open(self.server_certificate_chain, 'rb') as f2:
                certificate_chain = f2.read()
        return private_key, certificate_chain

    def set_certificate_chain(self, server_private_key, server_certificate_chain):
        # 判断设置的安全秘钥信息
        self.server_private_key = server_private_key
        self.server_certificate_chain = server_certificate_chain
        # 开始读取相关信息

    def _get_open_port(self):
        import socket
        s = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
        s.bind(("", 0))
        s.listen(1)
        port = s.getsockname()[1]
        s.close()
        return port

    # 可以写成装饰器的形式，也可以知己额调用 通过装饰器形式给我们的中间添加处理
    # middleware_class: type, **options: typing.Any
    def add_middleware(self, middleware_class: middlewareType, middleware_name=None, **options: typing.Any) -> None:
        # 修改middleware_type:type---middleware_type: typing.Callable
        # assert middleware_type in (BaseMiddleware)
        # print("是否子类：",issubclass(middleware_type,BaseMiddleware))
        # 断言检测，传入必须是一个可以调用是可调用的---比如方法或类对象等实现了,对于函数、方法、lambda 函式、 类以及实现了 __call__ 方法的类实例, 它都返回 True
        assert inspect.isclass(middleware_class), "当前仅支持类模式注册中间件！"
        assert callable(middleware_class), '当前必须是一个调用对象！'
        # 在第一个位置插入---名称如果有传入自定义的middleware_name，则使用自定义，不然默认就是类名
        curr_middleware_type_name = middleware_name or middleware_class.__name__
        # 断言服务去重核验
        assert curr_middleware_type_name not in [name.__class__.__name__ for name in self.user_middlewares], 'middleware repeat！'
        # 判断是否存在某个组合app属性存在
        # 进行中间件的实例化，需要传递app引用
        # 因为传入了app的引用，初始化的时候，也需要传递一下=========app实例- 兼容需要APP和不需要
        # fastapi只先统一体安装及到一个封装好的里面再进行统一的进行实例化，我是直接的把实例化的加到立碑里面了！
        self.user_middlewares.insert(0, middleware_class(self) if hasattr(middleware_class, 'composeapp_type') else middleware_class())

    # 添加事件信号处理器
    def add_expand_to_state(self, expand_type: typing.Type[BaseExpand], expand_name=None) -> None:
        # 根据类型来识别添加不懂的函数对象
        for b in expand_type.__bases__:
            assert b.__name__.endswith('BaseExpand'), '当前仅支持集成来自BaseExpand的扩展！'

        # 获取当前扩展的名称
        curr_expand_type_name = expand_name if expand_name else expand_type.__name__
        # 断言是否存在添加
        assert curr_expand_type_name not in [name for name in self.user_expands.keys()], 'expand add repeat！'
        # 初始化扩展实例对象
        self.user_expands[curr_expand_type_name] = expand_type()
        # 透传APP对象--默认建议所有的扩展都透传APP实例，主要是为了存储相关的信息到我们的app-statc里面保存
        self.user_expands[curr_expand_type_name].init_app(self)
        # 设置第三的的扩展存贮到state
        self.state.setattr(curr_expand_type_name, self.user_expands[curr_expand_type_name])

    # 可以写成装饰器的形式，也可以知己额调用 通过装饰器形式给我们的中间添加处理
    def add_servicer(self, service_cls: servericType) -> None:
        # assert middleware_type in (BaseMiddleware)
        # 断言必须是类，不能是一个函数
        assert inspect.isclass(service_cls), "当前仅支持类模式注册服务！"
        # assert not isinstance(service_cls,FunctionType),"当前仅支持类模式注册服务！"
        assert callable(service_cls), '当前必须是一个调用对象！'
        # 类型的名称
        curr_servicer_name = service_cls.__name__
        # 断言服务去重核验，避免重复注册
        assert curr_servicer_name not in [name for name in self.user_servicers.keys()], 'servicer repeat！'
        # 活动获取对应的类似的：add_GreeterServicer_to_server的方法函数
        add_func = self._get_servicer_add_func(service_cls)
        # add_func =add_GreeterServicer_to_server
        # 字典保持对应的服务名称和对应的调用的方法还要对应的服务对象
        # 保存一个元组到我们的字典里面
        self.user_servicers[curr_servicer_name] = (add_func, service_cls)

    def add_servicer_to_server(self, service_cls):
        """Register the given service class to the hello_srv."""
        # 关键是warnings.warn("some_old_function is deprecated", DeprecationWarning)这行代码被PyCharm识别到了。
        # 注意，上面只是让IDE能达到识别废弃并使用删除线提示而已
        warnings.warn("该方法和上面重复！可以不用了，这个主要是通过类找到对应的模块信息", DeprecationWarning)
        from importlib import import_module
        # proto_cls = service_cls.mro()[-2]
        # print(proto_cls)
        for b in service_cls.__bases__:
            # b.__name__=GreeterServicer
            if b.__name__.endswith('Servicer'):
                # inspect模块也被称为 检查现场对象。这里的重点在于“现场”二字，也就是当前运行的状态。
                # inspect模块提供了一些函数来了解现场对象，包括 模块、类、实例、函数和方法。
                proto_module = import_module(b.__module__)

    # 批量进行模块下的服务的导入
    def load_servicers_handlers_in_module(self, module):
        # 批量进行模块下的导入
        if isinstance(module, str):
            # 如果传去的是模块的目录，需要转换处理下
            from grpcframe.utils.import_string import import_string
            module = import_string(module)
            # 判断模块是否存在某个属性值
            assert module.__file__.endswith('__init__.py') and hasattr(module, 'auto'), '批量导入需要__init__模块包含auto自动加载方法调用：' \
                                                                                        '示例如：# from grpcframe.utils.import_class import auto_and_createVar as auto' \
                                                                                        '执行调用：auto(pro_path,locals())'

        # 变量模块当前的所属的属性信息
        for _, _servicer in inspect.getmembers(module, inspect.isclass):
            # 获取当前服务模块的名称
            curr_servicer_name = _servicer.__name__
            # 对于如果是直接导入了具体模块下类的情况：example.handlers.ser1，则直接进行处理即可,但这情况需要排除其他模块处理
            if curr_servicer_name.endswith('Servicer'):
                add_func = self._get_servicer_add_func(_servicer)
                self.user_servicers[curr_servicer_name] = (add_func, _servicer)
            # 如果是导入的整个外层的模块下的需要再执行一次变量
            elif curr_servicer_name not in [name for name in self.user_servicers.keys()]:
                add_func = self._get_servicer_add_func(_servicer)
                # 添加到服务列表中
                self.user_servicers[curr_servicer_name] = (add_func, _servicer)
        return self.user_servicers

    def _get_servicer_add_func(self, servicer: servericType):
        # 修改ervicer:type---servicer:typing.Callable
        for b in servicer.__bases__:
            # b.__name__=GreeterServicer
            if b.__name__.endswith('Servicer'):
                # inspect模块也被称为 检查现场对象。这里的重点在于“现场”二字，也就是当前运行的状态。
                # inspect模块提供了一些函数来了解现场对象，包括 模块、类、实例、函数和方法。
                # inspect.getmodule(object)：返回object所属的模块名
                m = inspect.getmodule(b)
                # 判断模块存在example.protos.hello_pb2_grpc.GreeterServicer--,add_GreeterServicer_to_server 这个的方法
                return getattr(m, f'add_{b.__name__}_to_server')

    def _get_reload_monitor(self):
        # 用于服务重启监听---目前这个还不行！！有待继续改进！
        import time
        from watchdog.observers import Observer
        from watchdog.events import FileSystemEventHandler
        from grpcframe.utils.asynctask import task
        import pathlib
        class ActionHandler(FileSystemEventHandler):
            def on_any_event(self, event):
                print('文件有变化:', event.event_type, event.src_path)
                # 发送自定义信号用户接收重启事件处理
                signal_send('reload', source='ThirdInternalErrorException', traceback='服务重启！！暂时没实现！')
                # 直接的结束了整个进程
                import os
                python = sys.executable
                os.execl(python, python, *sys.argv)

        print('当前监控的文件目录为：', pathlib.Path.cwd())

        @task(daemon=True)
        def ReloadHandler():
            event_handler = ActionHandler()
            observer = Observer()
            observer.schedule(event_handler, path='.', recursive=True)
            observer.start()
            try:
                while True:
                    time.sleep(1)
            except KeyboardInterrupt:
                observer.stop()
            observer.join()

        return ReloadHandler()

    @lru_cache()
    def get_app_settings(self):
        return AppSettings()

    def _restart_serve(self, signum, frame):
        self.shutdown()
        self.fastgrpc_server.stop(self.grpc_grace)
        time.sleep(self.grpc_grace or 1)
        # self._stopped = True
        self.run_serve()
        # raise KeyboardInterrupt

    def _stop_serve(self, signum, frame):
        # 标记进程停止
        self._stopped = True
        # 调用关闭结束回调
        self.shutdown()
        # 服务停止
        self.fastgrpc_server.stop(self.grpc_grace)
        # 进行结束休眠时间间隔
        time.sleep(self.grpc_grace or 1)
        # 进程退出
        sys.exit(0)
        # raise KeyboardInterrupt

    # ===============事件处理模块机制===============
    def startup(self) -> None:
        for handler in self.on_startup:
            handler()

    def shutdown(self) -> None:
        for handler in self.on_shutdown:
            handler()

    # 通过装饰器添加事件处理
    def on_event(self, event_type: str) -> typing.Callable:
        def decorator(func: typing.Callable) -> typing.Callable:
            self.add_event_handler(event_type, func)
            return func
        return decorator

    # 添加事件信号处理器
    def add_event_handler(self, event_type: str, func: typing.Callable) -> None:
        # 根据类型来识别添加不懂的函数对象
        assert event_type in ("startup", "shutdown")
        if event_type == "startup":
            self.on_startup.append(func)
        else:
            self.on_shutdown.append(func)

    def _register_signal(self):
        # windo只支持这个平台
        if platform.system().lower() == 'windows':
            signal.signal(signal.SIGINT, self._stop_serve)
        elif platform.system().lower() == 'linux':
            # 注册相关的事件
            signal.signal(signal.SIGINT, self._stop_serve)
            signal.signal(signal.SIGHUP, self._stop_serve)
            signal.signal(signal.SIGTERM, self._stop_serve)
            signal.signal(signal.SIGQUIT, self._stop_serve)

    def _make_server(self):
        compression = grpc.Compression.Gzip
        # 拦截器----判断是否是使用协程的方式启动服务
        if self.is_async:
            # 创建时间循环
            self._loop = asyncio.new_event_loop()
            # 设置当前使用的事件循环
            asyncio.set_event_loop(self._loop)
            self.fastgrpc_server = grpc.aio.server(
                migration_thread_pool=futures.ThreadPoolExecutor(max_workers=self._max_thread_pool_workers),
                compression=compression,
                interceptors=self.user_middlewares,
                # options=options
            )
        else:
            self.fastgrpc_server = grpc.server(futures.ThreadPoolExecutor(max_workers=self._max_thread_pool_workers),
                                               compression=compression,
                                               interceptors=self.user_middlewares)
        # 判断是否启用了安全信息
        # print('服务启动：', self.address)
        self.logger.info(f"服务启动：>>{self.address}")
        if self.is_certificate:
            private_key, certificate_chain = self.get_certificate_chain()
            server_credentials = grpc.ssl_server_credentials(((private_key, certificate_chain,),))
            self.fastgrpc_server.add_insecure_port(self.address, server_credentials)
        else:
            self.fastgrpc_server.add_insecure_port(self.address)

    def _check_servicers(self):
        pass
        if self.is_check_servicers:
            # >>> assert True     # 条件为 true 正常执行
            # >>> assert False    # 条件为 false 触发异常
            # assert self.user_servicers != {}, "请先添加对应服务对象!"
            # warnings.simplefilter('always', category=Warning, lineno=0, append=False)
            # warnings.warn("还没有注册相关服务！")
            assert len(self.user_servicers) > 0, '还没有注册相关服务！'

    def _print_servicers(self):
        if self.debug:
            self.logger.info(">>>>>>>>>>>>欢迎使用FastGrpcApp>>>>>>>>>>>>")
            for middleware_name in self.user_middlewares:
                self.logger.info(f"注册中间件名：>>{middleware_name.__class__.__name__}")
            self.logger.info(f">>>>>>" * 5)
            for servicer_name in self.user_servicers.keys():
                self.logger.info(f"注册服务类名：>>{servicer_name}")
            self.logger.info(f">>>>>>" * 5)
            for expand_name in self.user_expands.keys():
                self.logger.info(f"注册扩展：>>{expand_name}")

    def _register_serve(self):
        # 循环的方式注册我们的服务-注册我们的当前RPC的服务
        for name, (add_func, servicer) in self.user_servicers.items():
            # 服务注册时候传递当app实例------------------app实例-
            # 因为传入了app的引用，初始化的时候，也需要传递一下=========app实例-不需要的话则不传-兼容需要APP和不需要
            add_func(servicer(self) if hasattr(servicer, 'composeapp_type') else servicer(), self.fastgrpc_server)

    def _register_default_middlewares(self):
        # 默认注册一些中间件信息，如上下文处理中间件
        # PS: 关于中间件注册，最后注册，先执行，遵循的是洋葱模型
        from grpcframe.middlewares.grequest import GrequestMiddleware
        self.add_middleware(GrequestMiddleware)
        # 实现不同的方式处理
        from grpcframe.middlewares.flaskrequest import FlaskRequestMiddleware
        self.add_middleware(FlaskRequestMiddleware)
        # 注册日志追踪中间件
        if self.is_tracer:
            from grpcframe.middlewares.tracer import Zyxopentracing
            self.add_middleware(Zyxopentracing)

    def _register_default_handler(self):
        # 默认添加相关一些基础的服务处理，如健康检查服务
        from grpcframe.handlers.grpcio_health_checking.v1.health import HealthServicer
        self.add_servicer(HealthServicer)
        pass

    def _register_center_handler(self):
        # 默认添加相关一些基础的服务处理，如健康检查服务
        if self.register_center:
            # 调用注册中心的基类中响应的方法进行注册处理
            self.register_center.register(address=self._address_host, port=self.port)

    def _unregister_center_handler(self):
        # 默认添加相关一些基础的服务处理，如健康检查服务
        if self.register_center:
            # 调用注册中心的基类中响应的方法进行注册处理
            self.register_center.deregister()

    def _start(self):
        if self.fastgrpc_server:
            if self.is_async:
                # 使用安全协程的方式启动
                asyncio.run_coroutine_threadsafe(self.async_run(), self._loop)
            else:
                # 启动服务
                self.fastgrpc_server.start()

    def _wait_for_termination(self, timeout: typing.Any = None):
        if self.fastgrpc_server:
            if self.is_async:
                # 使用run_until_complete的方式来处理同步的方式
                return self._loop.run_until_complete(self.fastgrpc_server.wait_for_termination(timeout=timeout))
            else:
                return self.fastgrpc_server.wait_for_termination(timeout=timeout)

    def _stop_serve_callback(self):
        # 注销注册中心
        self._unregister_center_handler()

    def _run_forever(self, workers=1):
        # 先判断我们的服务列表是否为空为空则不启动！
        # ExitStack 实现类似GO 延迟执行的defer的延迟调用
        with contextlib.ExitStack() as stack:
            # 结束的时候回调处理
            stack.callback(self._stop_serve_callback)
            # 检测注册的服务
            self._check_servicers()
            # 发生启动信号事件处理
            self._register_signal()
            # 注册注册中间件默认名的--需要再_make_server之前调用
            self._register_default_middlewares()
            # 注册注册健康检测--需要再_make_server之前调用
            self._register_default_handler()
            # 打印注册的服务信息
            self._print_servicers()
            # 生成服务对象
            self._make_server()
            # 执行服务列表注册
            self._register_serve()
            # 执行注册的服务中心
            self._register_center_handler()
            # 应用启动
            self._start()
            # 调用应用启动回调处理
            self.startup()
            # 循环监听处理
            self._wait_for_termination()

    def run_serve(self, workers=1):
        if workers <= 1:
            # 正式的启动服务
            self._run_forever()
        else:
            sys.stdout.flush()
            freeze_support()
            workers_list = []
            for _ in range(workers):
                worker = multiprocessing.Process(target=self._run_forever, )
                worker.start()
                workers_list.append(worker)
            for worker in workers_list:
                worker.join()

    def _route(self, method: typing.Optional[str] = None, service: typing.Optional[str] = None) -> typing.Callable:
        def decorator(func: typing.Callable) -> typing.Callable:
            # 如果服务名称为空，需要报错失败异常！
            assert service, "对应方法的服务名称不能为空！"
            return func

        return decorator

    async def async_run(self):
        return await self.fastgrpc_server.start()

    async def async_wait_for_termination(self, timeout):
        return await self.fastgrpc_server.wait_for_termination(timeout=timeout)

    async def _shutdown(self):
        logging.info("正在关闭所有ss hello_srv")
        tasks = [t for t in asyncio.all_tasks() if t is not asyncio.current_task()]
        [task.cancel() for task in tasks]
        if self.fastgrpc_server:
            self.fastgrpc_server.close()
            logging.info(f"grpc hello_srv closed!")
        #
        self._loop.stop()
